package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import com.hexinfo.dmpro.common.model.MessageCenter;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dto.RuleToDistCPDTO;
import com.hexinfo.dmpro.sparing.dto.StateDataDTO;
import com.hexinfo.dmpro.sparing.dto.TableDistCPDTO;
import com.hexinfo.dmpro.sparing.model.HdfsHeatBackup;
import com.hexinfo.dmpro.sparing.model.HotStandbyLogSource;
import com.hexinfo.dmpro.sparing.model.HotStandbyLogTable;
import com.hexinfo.dmpro.sparing.service.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.File;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

/**
 * 有源分区字段热备方法
 */
@Service
@Slf4j
public class BackupSourceServiceImpl implements BackupSourceService {
    @Autowired
    private DataFilterService dataFilterService;
    @Autowired
    private DistCPService distCPService;
    @Autowired
    private LsFilterSouService lsFilterSouService;
    @Autowired
    private DataRefreshService dataRefreshService;
    @Autowired
    private MessageCenterService messageCenterService;
    @Autowired
    private DateDataService dateDataService;
    @Autowired
    private HotStandbyLogSourceService hotStandbyLogSourceService;
    @Autowired
    private UatSyncLogsTaskService uatSyncLogsTaskService;
    @Autowired
    private FutureService futureService;
    @Autowired
    private HotStandbyLogTableService hotStandbyLogTableService;

    @Value("${threadMonitoring}")
    private Integer threadMonitoring;

    //记录任务执行是否有异常
    private Map<String, Boolean> taskStateMap = new ConcurrentHashMap<String, Boolean>();
    //修改结束状态
    @Override
    public synchronized void updateTaskState(String souName,Boolean state) {
        taskStateMap.put(souName, state);
    }
    //获取结束状态
    public Boolean getTaskState(String souName) {
        return taskStateMap.get(souName);
    }
    //删除key
    public synchronized void deleteTaskState(String souName) {
        taskStateMap.remove(souName);
    }

    @Override
    public boolean execShell(RuleToDistCPDTO ruleToDistCPDTO,String logId) {
        //源级记录表添加消息中心id
        HotStandbyLogSource hotStandbyLogSource = hotStandbyLogSourceService.getById(logId);
        hotStandbyLogSource.setMessageId(ruleToDistCPDTO.getMessageId());
        hotStandbyLogSourceService.updateById(hotStandbyLogSource);
        //日志分析记录修改--用时详情修改
        hotStandbyLogSourceService.upHotLog(logId, DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"\n");
        String sourceDataName = ruleToDistCPDTO.getSourceDataName();  //同步的源名
        String backupSort = ruleToDistCPDTO.getBackupSort();    //1：有源分区字段，2：无源分区字段（时间窗口）
        String logPath = dataFilterService.path(sourceDataName);      //获取总日志路径
        updateTaskState(sourceDataName,true);
        //插入日志文件缓存操作
        FileAppender heatlog = new FileAppender(new File(logPath), 100, true);
        heatlog.append(logTime("上次热备时间："+ruleToDistCPDTO.getDate()+"--有无源类型："+ruleToDistCPDTO.getBackupSort()));
        heatlog.append(logTime("【"+sourceDataName+"】热备数据开始"));
        heatlog.flush();
        messageCenterChangeLog(ruleToDistCPDTO.getMessageId(),logPath);

        //1、把List<hdfs路径>整理成表级
        Map<String, List<HdfsHeatBackup>> entrys = dataFilterService.dbTableNameSort(ruleToDistCPDTO.getHdfsHeatBackups());
        heatlog.append(logTime("热备表数量："+entrys.size()+"，hdfs文件总数量："+ruleToDistCPDTO.getHdfsHeatBackups().size()));

        //日志分析记录修改--用时详情修改、同步范围起始时间
        upLSSyncRange(logId,ruleToDistCPDTO.getDate());

        heatlog.append(logTime("子线程超时时长（秒）："+threadMonitoring));
        heatlog.append(logTime("--------------------------------① LS过滤----------------------------------------"));
        //------------------------------① LS过滤+数据同步-----------------------------------
        //LS操作结束标志给初始值
        lsFilterSouService.updateLsStatusMap(sourceDataName,ScanCommonConstants.ClusterName.RUNNING.value);
        //方案1：异步方法，先LS结束后热备（方案2：按表LS+热备+比对）
        List<String> dataList = souAsyncFirstLs(entrys, ruleToDistCPDTO, heatlog, logId);
        //------------------------------② 判断是否不需要热备--------------------------------
        boolean dropState = true;
        //ls过滤后，表数量不为0，进行热备
        if (lsFilterSouService.getLsStatus(sourceDataName).equals(ScanCommonConstants.ClusterName.CONTINUE.value)){
            //日志分析记录修改--用时详情修改
            hotStandbyLogSourceService.upHotLog(logId, DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"\n"+
                            "【刷目标表】"+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"--");
            heatlog.append(logTime("--------------------------------③ 刷新目标hive、impala表--------------------------"));
            //------------------------------④ 刷新hive+刷新impala------------------------------------
            List<String> strings = lsFilterSouService.dataRefresh(sourceDataName, backupSort, heatlog);

            //日志分析记录修改--用时详情修改
            hotStandbyLogSourceService.upHotLog(logId, DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"\n"+
                    "【同步删除】"+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"--");

            heatlog.append(logTime("--------------------------------④ 同步删除----------------------------------------"));
            StateDataDTO stateDataDTO = dateDataService.dropDateData(sourceDataName, heatlog, ruleToDistCPDTO.getWhereScriptDTOS());
            dropState = stateDataDTO.isState();
            //日志分析记录修改--用时详情修改
            hotStandbyLogSourceService.upHotLog(logId, DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"\n"+
                    "【数据比对】"+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"--");

            heatlog.append(logTime("--------------------------------⑤ 数据比对----------------------------------------"));
            heatlog.flush();
            lsFilterSouService.dataContrast(sourceDataName,heatlog,ruleToDistCPDTO.getWhereScriptDTOS());
            //日志分析记录修改--用时详情修改
            hotStandbyLogSourceService.upHotLog(logId, DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss"));
        }
        lsFilterSouService.delLsStatus(sourceDataName);    //删除是否同步的key

        heatlog.append(logTime("--------------------------------数据热备结束----------------------------------------"));
        heatlog.append("\n\n"+"==========================================================================================="+"\n\n");
        heatlog.flush();
        //消息中心状态修改
        if(getTaskState(sourceDataName) && dropState){
            messageCenterChange(ruleToDistCPDTO.getMessageId(),ScanCommonConstants.ClusterName.YWC.value,logPath);
            deleteTaskState(sourceDataName);
            hotStandbyLogSourceService.upTaskState(logId,ScanCommonConstants.ClusterName.YWC.value);
            return true;
        }else {
            //修改消息中心状态
            List<HotStandbyLogTable> hotStandbyLogTablesMess = hotStandbyLogTableService.queryLikeList(logId, ScanCommonConstants.ClusterName.TBSB.value);
            if (hotStandbyLogTablesMess == null || hotStandbyLogTablesMess.size() == 0){
                messageCenterChange(ruleToDistCPDTO.getMessageId(),ScanCommonConstants.ClusterName.YWC.value,logPath);
                deleteTaskState(sourceDataName);
            }else {
                messageCenterChange(ruleToDistCPDTO.getMessageId(),ScanCommonConstants.ClusterName.YC.value,logPath);
                deleteTaskState(sourceDataName);
            }
            //修改日志状态
            List<HotStandbyLogTable> hotStandbyLogTables = hotStandbyLogTableService.queryLikeList(logId, ScanCommonConstants.ClusterName.SB.value);
            if (hotStandbyLogTables == null || hotStandbyLogTables.size() == 0){
                hotStandbyLogSourceService.upTaskState(logId,ScanCommonConstants.ClusterName.YWC.value);
                return true;
            }else {
                hotStandbyLogSourceService.upTaskState(logId,ScanCommonConstants.ClusterName.SB.value);
                return false;
            }
        }
    }

    /**
     * 日志分析记录修改--用时详情修改、同步范围起始时间
     */
    private void upLSSyncRange(String logId,Date lastDate) {
        Date date = DateUtil.date();
        String str = "【时间过滤】" + DateUtil.format(date, "yyyy-MM-dd HH:mm:ss")+"--";
        //上次热备时间向前偏移1小时
        String syncRangeStart = DateUtil.format(DateUtil.offsetHour(lastDate, -1), "yyyy-MM-dd HH:mm:ss");
        String syncRangeEnd = DateUtil.format(date, "yyyy-MM-dd HH:mm:ss");
        hotStandbyLogSourceService.upLSSyncRange(logId, syncRangeStart, syncRangeEnd,str);
    }

    /**
     * 万国到UAT
     * @param ruleToDistCPDTO
     * @return
     */
    @Override
    public String execShellWgToUat(RuleToDistCPDTO ruleToDistCPDTO) {
        //distcp
        String str = "";
        //封装获得每个表的hdfs实体类
        log.warn("======================封装获得每个表的hdfs实体类");
        TableDistCPDTO tableDistCPDTO = dataFilterService.getTableDistCPDTO(ruleToDistCPDTO, ruleToDistCPDTO.getHdfsHeatBackups());
        //获取对应库表集合
//        String libraryTableName = "";
//        for (HdfsHeatBackup hdfsHeatBackup : tableDistCPDTO.getHdfsHeatBackups()){
//            libraryTableName += hdfsHeatBackup.getSourceName()+",\n";
//        }
//        tableDistCPDTO.getUatSyncLogsTask().setLibraryTableName(libraryTableName);
//        uatSyncLogsTaskService.updateById(tableDistCPDTO.getUatSyncLogsTask());
        //调用热备方法
        log.warn("======================调用热备方法");
        str = lsFilterSouService.allLsFilterWgToUat(tableDistCPDTO);
        return str;
    }

    /**
     * 异步方法，先LS结束后热备
     */
    private List<String> souAsyncFirstLs(Map<String, List<HdfsHeatBackup>> entrys,
                              RuleToDistCPDTO ruleToDistCPDTO,FileAppender heatlog,String logId) {
        List<TableDistCPDTO> tableDistCPDTOs = new ArrayList<>();
        List<TableDistCPDTO> tableDistCPDTOS = Collections.synchronizedList(tableDistCPDTOs);
        //走全量的表集合
        List<TableDistCPDTO> wholeTableDistCPDTOs = new ArrayList<>();
        List<TableDistCPDTO> wholeTableDistCPDTOS = Collections.synchronizedList(wholeTableDistCPDTOs);
        Map<String,Future<Boolean>> futureMap = new HashMap<>();
        for (Map.Entry<String, List<HdfsHeatBackup>> entry : entrys.entrySet()) {
            //封装获得每个表的hdfs实体类
            TableDistCPDTO tableDistCPDTO = dataFilterService.getTableDistCPDTO(ruleToDistCPDTO, entry.getValue());
            //表日志文件缓存
            FileAppender tableLog = new FileAppender(new File(tableDistCPDTO.getLogPath()), 100, true);
            tableLog.append("==========================================================================================="+"\n\n");
            tableLog.append(logTime("-------------------------【LS过滤开始】"));
            //调用异步方法
            Future<Boolean> booleanFuture = lsFilterSouService.allLsFilter(tableDistCPDTOS, tableDistCPDTO, heatlog, tableLog, wholeTableDistCPDTOS);
            futureMap.put(entry.getKey(),booleanFuture);
        }
        //判断LS全部结束
        StateDataDTO stateDataDTO = futureService.futureStatusMap(futureMap, heatlog,"LS过滤");
        List<String> dataList = stateDataDTO.getDataList();
        heatlog.append(logTime("-------------------------【LS过滤结束】"));
        heatlog.flush();
        //数据备份同步
        List<String> strings = lsFilterSouService.backupData(tableDistCPDTOS, ruleToDistCPDTO.getSourceDataName(),
                heatlog, logId, wholeTableDistCPDTOS, ruleToDistCPDTO.getSyncRuleConfSet());
        heatlog.flush();
        return dataList;

    }

    /**
     * 修改这一次消息中心日志路径
     * @return
     */
    private void messageCenterChangeLog(String id,String logPath) {
        MessageCenter mess = messageCenterService.getById(id);
        mess.setLogPath(logPath);
        messageCenterService.updateById(mess);
    }

    /**
     * 修改这一次消息中心状态为已完成
     * @return
     */
    private void messageCenterChange(String id,String state,String logPath) {
        MessageCenter mess = messageCenterService.getById(id);
        mess.setDataType(state);
        messageCenterService.updateById(mess);
    }

    /**
     * 日志加日期时间
     * @param str
     * @return
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
